#pragma once

#include <rocksdb/comparator.h>
#include <rocksdb/slice.h>
#include <rocksdb/status.h>

#include <iostream>
#include <string>

#include "slash/coding.h"

class ZSetsComparator : public rocksdb::Comparator {
 public:
  const char* Name() const override { return "redis.ZSetsComparator"; }

  /*
   *  |  <Key Size>  |      <Key>      |  <Version>  |  <Score>  | <Member> |
   *       4 Bytes     Key Size Bytes      4 Bytes      8 Bytes
   */

  int Compare(const rocksdb::Slice& a, const rocksdb::Slice& b) const override {
    assert(a.size() > sizeof(int32_t));
    assert(a.size() >= slash::DecodeFixed32(a.data()) + 2 * sizeof(int32_t) +
                           sizeof(uint64_t));
    assert(b.size() > sizeof(int32_t));
    assert(b.size() >= slash::DecodeFixed32(b.data()) + 2 * sizeof(int32_t) +
                           sizeof(uint64_t));

    const char* a_ptr = a.data();
    const char* b_ptr = b.data();

    int32_t a_zset_key_len = slash::DecodeFixed32(a_ptr);
    int32_t b_zset_key_len = slash::DecodeFixed32(b_ptr);

    rocksdb::Slice a_zset_key_prefix(a_ptr,
                                     a_zset_key_len + 2 * sizeof(int32_t));
    rocksdb::Slice b_zset_key_prefix(b_ptr,
                                     b_zset_key_len + 2 * sizeof(int32_t));

    a_ptr += a_zset_key_len + 2 * sizeof(int32_t);
    b_ptr += b_zset_key_len + 2 * sizeof(int32_t);

    int ret = a_zset_key_prefix.compare(b_zset_key_prefix);
    if (ret) {
      return ret;
    }

    double a_zset_score = slash::DecodeDouble(a_ptr);
    double b_zset_score = slash::DecodeDouble(b_ptr);

    if (a_zset_score != b_zset_score) {
      return a_zset_score < b_zset_score ? -1 : 1;
    } else {
      a_ptr = a_ptr + sizeof(uint64_t);
      b_ptr = b_ptr + sizeof(uint64_t);
      if (a_ptr - a.data() == a.size() && b_ptr - b.data() == b.size()) {
        return 0;
      } else if (a_ptr - a.data() == a.size()) {
        return -1;
      } else if (b_ptr - b.data() == b.size()) {
        return 1;
      } else {
        rocksdb::Slice a_zset_member(a_ptr, a.size() - (a_ptr - a.data()));
        rocksdb::Slice b_zset_member(b_ptr, b.size() - (b_ptr - b.data()));
        ret = a_zset_member.compare(b_zset_member);
        if (ret) {
          return ret;
        }
      }
    }
    return 0;
  }

  bool Equal(const rocksdb::Slice& a, const rocksdb::Slice& b) const override {
    return !Compare(a, b);
  }

  void FindShortestSeparator(std::string* start,
                             const rocksdb::Slice& limit) const override {}

  void FindShortSuccessor(std::string* key) const override {}

 private:
};
